package com.hefei.garden.config.kafka.config;

import com.google.common.collect.Maps;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.RangeAssignor;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.Collections;
import java.util.Map;

/**
 * @author: devil
 * @Date: 2022/8/21 16
 * @Description: 消费者配置
 */
@Configuration
@RefreshScope
public class KafkaConsumerConfig {

//    @Value("${spring.kafka.servers}")
//    private String bootstrapServers;
//    @Value("${spring.kafka.consumer.enableAutoCommit}")
//    private boolean enableAutoCommit;
//    @Value("${spring.kafka.consumer.fetchMinBytes}")
//    private int fetchMinBytes;
//    @Value("${spring.kafka.consumer.fetchMaxWaitMs}")
//    private int fetchMaxWaitMs;
//    @Value("${spring.kafka.consumer.metadataMaxAgeMs}")
//    private int metadataMaxAgeMs;
//    @Value("${spring.kafka.consumer.concurrency}")
//    private int concurrency;
//    @Value("${spring.kafka.consumer.pollTimeout}")
//    private int pollTimeout;
//
//    @Bean("consumerFactory")
//    public ConsumerFactory<String, String> consumerFactory() {
//        Map<String, Object> propsMap = Maps.newHashMapWithExpectedSize(8);
//        propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//        propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
//        propsMap.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, Collections.singletonList(RangeAssignor.class));
//        propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//        propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
//        propsMap.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, fetchMinBytes);
//        propsMap.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, fetchMaxWaitMs);
//        propsMap.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, metadataMaxAgeMs);
//        return new DefaultKafkaConsumerFactory<>(propsMap);
//    }
//
//    @Bean("kafkaListenerContainerFactory")
//    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> getKafkaListenerContainerFactory(@Autowired ConsumerFactory<String, String> consumerFactory) {
//        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
//        factory.setConsumerFactory(consumerFactory);
//        // 机器数量 * concurrency <= 分区数
//        factory.setConcurrency(concurrency);
//        // 拉取不到消息，等待多少ms，一般设置2000毫秒左右
//        factory.getContainerProperties().setPollTimeout(pollTimeout);
//        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
//        // 是否支持批量消费
//        factory.setBatchListener(true);
//        return factory;
//    }

}
